package cn.monkey.io.transport.netty;

import cn.monkey.io.transport.core.Connection;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.ChannelGroupFutureListener;
import io.netty.util.AttributeKey;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.Objects;
import java.util.function.Predicate;

public class DefaultChannelGroupInbound implements ChannelGroupInbound {

    private final ChannelGroup channelGroup;

    private final Flux<Object> msgPublisher;

    public DefaultChannelGroupInbound(ChannelGroup channelGroup) {
        this.channelGroup = channelGroup;
        AttributeKey<ChannelConnection> connectionAttributeKey = AttributeKey.valueOf(Connection.KEY);
        List<Flux<Object>> fluxList = this.channelGroup.stream()
                .map(channel -> {
                    ChannelConnection channelConnection = channel.attr(connectionAttributeKey).get();
                    return Objects.requireNonNullElseGet(channelConnection, () -> new DefaultChannelConnection(channel).bind());
                })
                .map(c -> c.receive(Object.class))
                .toList();
        this.msgPublisher = Flux.merge(fluxList);
    }

    @Override
    public Mono<Void> close() {
        ChannelGroupFuture future = this.channelGroup.close();
        return Mono.create(sink -> future.addListener((ChannelGroupFutureListener) channelFutures -> {
            if (channelFutures.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(channelFutures.cause());
        }));
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.msgPublisher.filter(clazz::isInstance)
                .map(clazz::cast);
    }

    @Override
    public ChannelGroup channelGroup() {
        return this.channelGroup;
    }
}
